package spring.cloud.tasks.api.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.StopWatch;
import spring.cloud.tasks.api.database.entity.TaskConfigurationDto;
import spring.cloud.tasks.api.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.api.database.service.TaskConfigurationDtoService;
import spring.cloud.tasks.api.domain.different_info.DifferentType;
import spring.cloud.tasks.api.domain.different_info.TaskConfigurationDifferentInfo;
import spring.cloud.tasks.api.domain.different_info.TaskDifferentInfo;
import spring.cloud.tasks.api.domain.task.TaskConfiguration;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceZooKeeperClient;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperClusterService;
import spring.cloud.tasks.api.service.ZooKeeperDatabaseTaskDifferentInfoService;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.node.TaskConfigurationNode;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeperRepository;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Slf4j
public class ZooKeeperDatabaseTaskDifferentInfoServiceImpl implements ZooKeeperDatabaseTaskDifferentInfoService {


    private static final int DIFF_THREAD_NUM = 10;

    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;

    @Resource
    private TaskConfigurationDtoService taskConfigurationDtoService;

    @Resource
    private TaskService taskService;

    @Resource
    private ZooKeeperClusterService zooKeeperService;

    @Resource
    private ZooKeeperRepository zooKeeperRepository;

    private ExecutorService service;

    @PostConstruct
    public void init() {
        if (service != null) {
            service.shutdownNow();
        }
        service = Executors.newFixedThreadPool(DIFF_THREAD_NUM, new ThreadFactory("diff-zk-db-thread", false));
    }

    @PreDestroy
    public void destroy() {
        if (service != null) {
            service.shutdownNow();
        }
    }

    @Override
    public List<TaskDifferentInfo> differentByZooKeeperClusterKey(String zooKeeperClusterKey) throws RuntimeException {

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        List<String> nameSpaceList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceList(zooKeeperClusterKey);

        List<Callable<List<TaskDifferentInfo>>> callableList = new ArrayList<>();
        for (final String nameSpace : nameSpaceList) {
            Callable<List<TaskDifferentInfo>> callable = new Callable<List<TaskDifferentInfo>>() {
                @Override
                public List<TaskDifferentInfo> call() throws Exception {
                    return differentByNameSpace(nameSpace);
                }
            };
            callableList.add(callable);
        }

        List<TaskDifferentInfo> taskDifferentInfoList = new ArrayList<>();
        try {
            List<Future<List<TaskDifferentInfo>>> futureList = service.invokeAll(callableList);

            for (Future<List<TaskDifferentInfo>> future : futureList) {
                List<TaskDifferentInfo> taskDifferentInfos = future.get();
                if (taskDifferentInfos != null && !taskDifferentInfos.isEmpty()) {
                    taskDifferentInfoList.addAll(taskDifferentInfos);
                }
            }
        } catch (InterruptedException e) {// NOSONAR
            log.warn("the thread is interrupted", e);
            throw new RuntimeException("the diff thread is interrupted", e);
        } catch (Exception e) {
            log.error("exception happens during execute diff operation", e);
            throw new RuntimeException(e);
        }
        stopWatch.stop();
        log.info("Finish diff zkcluster:{}, which cost {}ms", zooKeeperClusterKey, stopWatch.getTotalTimeSeconds());

        return taskDifferentInfoList;
    }

    @Override
    public List<TaskDifferentInfo> differentByNameSpace(String nameSpace) throws RuntimeException {
        StopWatch stopWatch = new StopWatch();
        List<TaskDifferentInfo> taskDifferentInfoList = new ArrayList<>();
        try {
            stopWatch.start();
            List<TaskConfigurationDto> taskConfigurationDtoList = taskConfigurationDtoService.getTaskConfigurationDtoListByNameSpace(nameSpace);
            if (taskConfigurationDtoList == null || taskConfigurationDtoList.isEmpty()) {
                return taskDifferentInfoList;
            }
            ZooKeeper zooKeeper = initNameSpaceZooKeeper(nameSpace);
            if (zooKeeper == null) {
                return taskDifferentInfoList;
            }
            for (TaskConfigurationDto taskConfigurationDto : taskConfigurationDtoList) {
                String taskId = taskConfigurationDto.getTaskId();
                if (!checkExists(zooKeeper, taskId)) {
                    taskDifferentInfoList.add(new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ONLY, new ArrayList<>()));
                    continue;
                }
                TaskConfiguration taskConfigurationFromZooKeeper = TaskConfigurationAndZooKeeperUtil.getTaskConfigurationFromZooKeeper(zooKeeper, taskId);
                TaskDifferentInfo taskDifferentInfo = different(nameSpace, taskConfigurationDto, taskConfigurationFromZooKeeper);
                if (taskDifferentInfo != null) {
                    taskDifferentInfoList.add(taskDifferentInfo);
                }
            }
            Set<String> taskIdSetInDb = taskConfigurationDtoList.stream().map(taskConfigurationDto -> {
                return taskConfigurationDto.getTaskId();
            }).collect(Collectors.toSet());

            List<String> taskIdList = taskService.getTaskIdListFromZooKeeper(nameSpace);
            for (String taskId : taskIdList) {
                if (taskIdSetInDb == null || taskIdSetInDb.isEmpty() || !taskIdSetInDb.contains(taskId)) {
                    taskDifferentInfoList.add(new TaskDifferentInfo(nameSpace, taskId, DifferentType.ZOOKEEPER_ONLY, new ArrayList<>()));
                }
            }
        } catch (RuntimeException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Exception e) {
            log.error("exception throws during diff by nameSpace [{}]", nameSpace, e);
            throw new RuntimeException(e);
        } finally {
            stopWatch.stop();
            log.info("Finish diff nameSpace:{} which cost {}s", nameSpace, stopWatch.getTotalTimeSeconds());
        }

        return taskDifferentInfoList;
    }


    @Override
    public TaskDifferentInfo differentByTaskId(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper;
        try {
            zooKeeper = initNameSpaceZooKeeper(nameSpace);
            if (zooKeeper == null) {
                return null;
            }
            log.info("start to diff job:{}", taskId);

            TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
            TaskConfiguration taskConfigurationFromZooKeeper = TaskConfigurationAndZooKeeperUtil.getTaskConfigurationFromZooKeeper(zooKeeper, taskId);

            if (taskConfigurationDto == null) {
                if (taskConfigurationFromZooKeeper != null) {
                    return new TaskDifferentInfo(nameSpace, taskId, DifferentType.ZOOKEEPER_ONLY, new ArrayList<>());
                } else {
                    return null;
                }
            } else {
                if (taskConfigurationFromZooKeeper == null) {
                    return new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ONLY, new ArrayList<>());
                } else {
                    // diff only when dbJobConfig and zkJobConfig both not null
                    return different(nameSpace, taskConfigurationDto, taskConfigurationFromZooKeeper);
                }
            }
        } catch (Exception e) {
            log.error("exception throws during diff by nameSpace [{}] and job [{}]", nameSpace, taskId, e);
            throw new RuntimeException(e);
        }
    }

    private ZooKeeper initNameSpaceZooKeeper(String nameSpace) {
        NameSpaceConfiguration nameSpaceConfiguration = zooKeeperService.getNameSpaceConfiguration(nameSpace);
        if (nameSpaceConfiguration == null) {
            log.warn("NameSpace:" + nameSpace + "不存在");
            return null;
        }
        NameSpaceZooKeeperClient nameSpaceZooKeeperClient = zooKeeperService.getNameSpaceZooKeeperClient(nameSpace);
        if (nameSpaceZooKeeperClient != null && nameSpaceZooKeeperClient.isConnected()) {
            return zooKeeperRepository.getZooKeeper(nameSpaceZooKeeperClient.getCuratorFramework());
        }
        log.warn("连接ZooKeeper失败:" + nameSpace);
        return null;
    }

    private boolean checkExists(ZooKeeper zooKeeper, String taskId) {
        return zooKeeper.checkExists(TaskNodePath.getTaskNodePath(taskId));
    }

    /**
     * zk中的任务配置和db中的任务配置的对比。
     *
     * @param dbTaskConfiguration db里面的配置。
     * @param zkTaskConfiguration zk里面的配置
     */
    protected TaskDifferentInfo different(String nameSpace, TaskConfiguration dbTaskConfiguration, TaskConfiguration zkTaskConfiguration) {
        String taskId = dbTaskConfiguration.getTaskId();

        List<TaskConfigurationDifferentInfo> list = new ArrayList<>();

        different(TaskConfigurationNode.task_type, dbTaskConfiguration.getTaskType(), zkTaskConfiguration.getTaskType(), list);
        //task_id
        different(TaskConfigurationNode.xid, dbTaskConfiguration.getXId(), zkTaskConfiguration.getXId(), list);
        different(TaskConfigurationNode.description, dbTaskConfiguration.getDescription(), zkTaskConfiguration.getDescription(), list);
        //
        different(TaskConfigurationNode.cron_expression, dbTaskConfiguration.getCronExpression(), zkTaskConfiguration.getCronExpression(), list);
        different(TaskConfigurationNode.pause_period_date, dbTaskConfiguration.getPausePeriodDate(), zkTaskConfiguration.getPausePeriodDate(), list);
        different(TaskConfigurationNode.pause_period_time, dbTaskConfiguration.getPausePeriodTime(), zkTaskConfiguration.getPausePeriodTime(), list);
        //
        different(TaskConfigurationNode.timeout_seconds, String.valueOf(dbTaskConfiguration.getTimeoutSeconds()), String.valueOf(zkTaskConfiguration.getTimeoutSeconds()), list);
        different(TaskConfigurationNode.timeout_4_alarm_seconds, String.valueOf(dbTaskConfiguration.getTimeout4AlarmSeconds()), String.valueOf(zkTaskConfiguration.getTimeout4AlarmSeconds()), list);
        //
        //
        different(TaskConfigurationNode.task_item_total_count, String.valueOf(dbTaskConfiguration.getTaskItemTotalCount()), String.valueOf(zkTaskConfiguration.getTaskItemTotalCount()), list);
        different(TaskConfigurationNode.load_level, String.valueOf(dbTaskConfiguration.getLoadLevel()), String.valueOf(zkTaskConfiguration.getLoadLevel()), list);
        //
        different(TaskConfigurationNode.task_item_parameters, dbTaskConfiguration.getTaskItemParameters(), zkTaskConfiguration.getTaskItemParameters(), list);
        different(TaskConfigurationNode.task_parameter, dbTaskConfiguration.getTaskParameter(), zkTaskConfiguration.getTaskParameter(), list);
        //
        different(TaskNodePath.enabled, String.valueOf(dbTaskConfiguration.getEnabled()), String.valueOf(zkTaskConfiguration.getEnabled()), list);
        different(TaskNodePath.prefer_executor_id_list, dbTaskConfiguration.getPreferExecutorIdList(), zkTaskConfiguration.getPreferExecutorIdList(), list);
        if (!list.isEmpty()) {
            return new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ZOOKEEPER_HAS_DIFFERENCE, list);
        } else {
            return null;
        }
    }

    public void different(String key, String valueInDb, String valueInZk, List<TaskConfigurationDifferentInfo> taskConfigurationDifferentInfoList) {
        // 这里处理所有valueInDB 为空的情况
        if (StringUtils.isEmpty(valueInDb)) {
            if (StringUtils.isEmpty(valueInZk)) {
                return;
            } else {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            }

        } else {
            // valueInDB != null && valueInZk == null
            if (StringUtils.isEmpty(valueInZk)) {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            }
            // 处理非String类型
            if (!valueInDb.trim().equals(valueInZk.trim())) {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            } else {
                return;
            }
        }


    }


}
